/*
 * Copyright 2021 Shulie Technology, Co.Ltd
 * Email: shulie@shulie.io
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.shulie.surge.data.runtime.disruptor;

/**
 * {@link SequenceBarrier} handed out for gating {@link EventProcessor}s on a cursor sequence and optional dependent {@link EventProcessor}(s),
 * using the given WaitStrategy.
 */
final class ProcessingSequenceBarrier implements SequenceBarrier
{
	private final WaitStrategy waitStrategy;
	private final Sequence dependentSequence;
	private volatile boolean alerted = false;
	private final Sequence cursorSequence;
	private final Sequencer sequencer;

	public ProcessingSequenceBarrier(final Sequencer sequencer,
			final WaitStrategy waitStrategy,
			final Sequence cursorSequence,
			final Sequence[] dependentSequences)
	{
		this.sequencer = sequencer;
		this.waitStrategy = waitStrategy;
		this.cursorSequence = cursorSequence;
		if (0 == dependentSequences.length)
		{
			dependentSequence = cursorSequence;
		}
		else
		{
			dependentSequence = new FixedSequenceGroup(dependentSequences);
		}
	}

	@Override
	public long waitFor(final long sequence)
			throws AlertException, InterruptedException, TimeoutException
	{
		checkAlert();

		long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);

		if (availableSequence < sequence)
		{
			return availableSequence;
		}

		return sequencer.getHighestPublishedSequence(sequence, availableSequence);
	}

	@Override
	public long getCursor()
	{
		return dependentSequence.get();
	}

	@Override
	public boolean isAlerted()
	{
		return alerted;
	}

	@Override
	public void alert()
	{
		alerted = true;
		waitStrategy.signalAllWhenBlocking();
	}

	@Override
	public void clearAlert()
	{
		alerted = false;
	}

	@Override
	public void checkAlert() throws AlertException
	{
		if (alerted)
		{
			throw AlertException.INSTANCE;
		}
	}
}